package com.zdb.demo.flink.streaming;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Iterator;

/**
 * 创建一个 Flink Streaming Job
 *
 * 使用 EventTime
 * 使用 SlidingWindow，size = 10s  slide=3s
 * Watermark 使用 BoundedOutOfOrdernessTimestampExtractor，maxOutOfOrderness=2s
 * Allowed lateness = 2s
 * 截图证明
 *
 * 晚到在2s内的会被正确放进 window 内
 * 晚到超过2s的会被舍弃（不考虑 side output
 */
public class H5 {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration(){{
            setInteger("rest.port", 9191);
            setBoolean("local.start-webserver", true);
        }};
        final StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        streamEnv.setParallelism(4);
        streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<Tuple3<String, Long, String>> ds = streamEnv.socketTextStream("localhost", 9999).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value!=null && value.length() > 0;
            }
        }).map(new MapFunction<String, Tuple3<String, Long, Long>>() {
            @Override
            public Tuple3<String, Long, Long> map(String value) throws Exception {
                String[] strs = value.split(",");
                return new Tuple3<String, Long, Long>(strs[0], Long.parseLong(strs[1].trim()), Long.parseLong(strs[2].trim()) * 1000);
            }
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<String, Long, Long>>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(Tuple3<String, Long, Long> element) {
                return element.f2;
            }
        }).keyBy(0)
                .timeWindow(Time.seconds(10), Time.seconds(3)).allowedLateness(Time.seconds(2))
                .aggregate(new AggregateFunction<Tuple3<String, Long, Long>, Tuple2<String, Long>, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> createAccumulator() {
                        return null;
                    }

                    @Override
                    public Tuple2<String, Long> add(Tuple3<String, Long, Long> value, Tuple2<String, Long> accumulator) {
                        if (accumulator == null) {
                            return new Tuple2<>(value.f0, value.f1);
                        } else {
                            if (value.f1 > accumulator.f1) {
                                accumulator.f1 = value.f1;
                                return accumulator;
                            } else {
                                return accumulator;
                            }
                        }
                    }

                    @Override
                    public Tuple2<String, Long> getResult(Tuple2<String, Long> accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Tuple2<String, Long> merge(Tuple2<String, Long> a, Tuple2<String, Long> b) {
                        return null;
                    }
                }, new WindowFunction<Tuple2<String, Long>, Tuple3<String, Long, String>, Tuple, TimeWindow>() {
                    @Override
                    public void apply(Tuple tuple, TimeWindow window, Iterable<Tuple2<String, Long>> input, Collector<Tuple3<String, Long, String>> out) throws Exception {
                        Iterator<Tuple2<String, Long>> iter = input.iterator();
                        if(iter.hasNext()) {
                            Tuple2<String, Long> v = iter.next();
                            out.collect(new Tuple3<>(v.f0, v.f1, window.getStart() + " --> " + window.getEnd()));
                            while (iter.hasNext()) {
                                System.out.println("------- window function Iter has next:" + iter.next());
                            }
                        }
                    }
                });

        ds.print("max------>");

        streamEnv.execute("SideOutputDemo stream job");
    }

}
